package com.gugee.tiktok.data.consumer.blogger.core.main;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.tiktok.common.model.AwemeScore;
import com.gugee.tiktok.common.model.BloggerScore;
import com.gugee.tiktok.common.model.mysql.BloggerMain;
import com.gugee.tiktok.common.model.spider.blogger.BloggerResponse;
import com.gugee.tiktok.common.model.spider.blogger.User;
import com.gugee.tiktok.common.model.system.SystemInfo;
import com.gugee.common.exception.BizException;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.common.model.dal.blogger.BloggerMainRepository;
import com.gugee.tiktok.data.consumer.blogger.config.DataConsumerConfig;
import com.gugee.tiktok.data.consumer.blogger.core.service.AwemeClient;
import com.gugee.tiktok.data.consumer.blogger.core.spider.BloggerSpiderClient;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.client.ResourceAccessException;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-10-14 15:20
 */
@Slf4j
@Component
public class UpdateBloggerMain {

    final String updateBloggerMain = "UpdateBlogger-";
    final String updateBloggerMainOneThousand = "UpdateBlogger-top1000-";
    @Autowired
    AwemeClient awemeClient;
    @Autowired
    BloggerSpiderClient bloggerSpiderClient;
    @Autowired
    BloggerMainRepository bloggerMainRepository;
    @Autowired
    SaveBloggerRecordMain saveBloggerRecordMain;
    @Autowired
    BloggerMain.BloggerMainMapper bloggerMapper;
    @Autowired
    ScheduleTemplate scheduleTemplate;
    @Autowired
    SystemInfo systemInfo;
    @Autowired
    ESSaveBloggerMain esSaveBloggerMain;
    @Autowired
    DataConsumerConfig dataConsumerConfig;
    private ThreadPoolExecutor customPoolMain;

    @Autowired
    KafkaTemplate<Long, BloggerMessage> kafkaTemplateBlogger;
    @PostConstruct
    public void init() {
        log.info("UpdateBloggerMainEventListener  init");
        customPoolMain = scheduleTemplate.getCustomThreadPool(updateBloggerMain, dataConsumerConfig.getCorePollSize(), dataConsumerConfig.getCorePollSize(), 0, dataConsumerConfig.getQueueSize(), TimeUnit.MINUTES);
    }

    public void handle(List<BloggerMessage> bloggerMessages)   {
        if(bloggerMessages==null || bloggerMessages.size()==0){
            log.info("bloggerMessages is empty!");
            return;
        }
        List<BloggerMain> bloggerMains=Collections.synchronizedList(new ArrayList<>(bloggerMessages.size()));
        ThreadPoolExecutor customThreadPool=customPoolMain;
        Collection<Future<?>> futures = new LinkedList<>();

        final ThreadPoolExecutor customThreadPool2 = customThreadPool;
        bloggerMessages.forEach(
                bloggerMessage -> futures.add(customThreadPool2.submit(() -> {
                    try {
                        run(bloggerMessage, bloggerMains);

                    } catch(ResourceAccessException  e){
                        log.error(" 超时-----需重试{}次：{}" ,bloggerMessage.getRetryTimes()-1,e.getMessage());
                        if(bloggerMessage.getRetryTimes()>0){
//                            if(e.getMessage()!=null && e.getMessage().contains("SocketTimeoutException")){
//                                log.error(" 超时-----需重试{}次：{}" ,bloggerMessage.getRetryTimes()-1,e.getMessage());
//                            }
//                            if(e.getMessage()!=null && e.getMessage().contains("HttpHostConnectException") && e.getMessage().contains("Connection refused")){
//                                log.error(" 连接被拒绝需重试{}次：{}" ,bloggerMessage.getRetryTimes()-1,e.getMessage());
//                            }
                            bloggerMessage.setRetryTimes(bloggerMessage.getRetryTimes()-1);
                            kafkaTemplateBlogger.send(dataConsumerConfig.getBloggerOnlineTopic(),bloggerMessage);
                        }
                        if(bloggerMessage.getRetryTimes()==0){
                            log.info(" 重试{}次后依旧失败:uid={}" ,bloggerMessage.getRetryLimitTimes(),bloggerMessage.getUid());
                        }


                    }catch (BizException e){
                        log.error(" BizException-----需重试{}次：{}" ,bloggerMessage.getRetryTimes()-1,e.getMessage());
                        if(bloggerMessage.getRetryTimes()>0){
                            log.error(" {};需重试{}次" ,e.getMessage(),bloggerMessage.getRetryTimes()-1);
                            bloggerMessage.setRetryTimes(bloggerMessage.getRetryTimes()-1);
                            kafkaTemplateBlogger.send(dataConsumerConfig.getBloggerOnlineTopic(),bloggerMessage);
                        }
                        if(bloggerMessage.getRetryTimes()==0){
                            log.info(" 重试{}次后依旧失败:uid={}" ,bloggerMessage.getRetryLimitTimes(),bloggerMessage.getUid());

                        }
                    }
                }))
        );
        for (Future<?> future : futures) {
            try {
                future.get();
            }  catch (ResourceAccessException e){
                log.error(" ResourceAccessException.{}" ,e.getMessage());
            } catch(InterruptedException | ExecutionException e) {
                e.printStackTrace();
                log.error(" future .{}" ,e.getMessage());
            }
        }

   }
    private void run(BloggerMessage bloggerMessage,List<BloggerMain> bloggerMains) {
        long uid=bloggerMessage.getUid();
        String region=bloggerMessage.getRegion();
        String bloggerType=bloggerMessage.getBloggerType();
        long start = System.currentTimeMillis();

        //根据uid,region调用接口获取播主相关的数据
        BloggerResponse response = bloggerSpiderClient.get(uid, region);
        User user=response.getUser();
        if (user == null) {
            log.warn("UpdateBloggerMain bloggerSpiderClient response.getUser() is null ;uid:{},region:{},status_code:{},status_msg:{}", uid , region,response.getStatus_code(),response.getStatus_msg());
            bloggerMainRepository.updateBloggerStatusByoUid(new Date(),response.getStatus_code(),uid);
            return;
        }
        long bloggerSpiderTime=(System.currentTimeMillis()-start)/1000;
        BloggerMain blogger = bloggerMapper.mapper(user);
        //根据uid 采集接口获取更新视频，并更新||插入到mysql
        long awemeSpideUpdateTime=System.currentTimeMillis();
        awemeClient.refreshBloggerAweme(bloggerType,uid);
        awemeSpideUpdateTime=(System.currentTimeMillis() - awemeSpideUpdateTime) / 1000;
        long growthScoreTime=System.currentTimeMillis();
        //查询统计90天视频，平均值
        var bloggerAwemeExtend = awemeClient.getBloggerAwemeExtend(uid);
        blogger.setOAvgLikes(bloggerAwemeExtend.getAvgLikes());
        blogger.setOAvgReviews(bloggerAwemeExtend.getAvgReviews());
        blogger.setOAvgShare(bloggerAwemeExtend.getAvgShare());
        blogger.setOAvgViews(bloggerAwemeExtend.getAvgViews());
        if(!StringUtils.isEmpty(bloggerAwemeExtend.getAnchor())  && !bloggerAwemeExtend.getAnchor().equals("NONE")){
            blogger.setOWithFusionShopEntry(true);
            blogger.setWithFusionShopEntry(true);
        }

        //todo 查询播主当前已保存的最近90d视频，与下面更新分数处理重复查询
        val bloggerAwemes = awemeClient.getBloggerAwemes(blogger.getOUid());
        AwemeScore awemeScore = new AwemeScore(blogger, bloggerAwemes);


        //todo 再次查询播主当前已保存的最近90d视频,并更新Score,可优化为批量更新.此处两个refresh很花时间
        awemeClient.refreshBloggerAwemeScore(awemeScore);
        awemeClient.refreshBloggerAwemeForES(blogger.getOUid());

        blogger.setLastPubTime(new Date());
        // 计算博主得分
        BloggerScore.ScoreParms scoreParms = new BloggerScore.ScoreParms();
        scoreParms.fans = blogger.getOFollowerCount();
        scoreParms.artnumbers = bloggerAwemes.size();
        scoreParms.comment = (double) blogger.getOAvgReviews();
        scoreParms.share = (double) blogger.getOAvgShare();
        scoreParms.like = (double) blogger.getOAvgLikes();
        blogger.setScore(BloggerScore.GetTotalScore(scoreParms));

        //更新播主分数 todo 可以积累或发送到消息队列中批量处理
        var optionalBlogger = bloggerMainRepository.findByoUid(blogger.getOUid());
        var origin = optionalBlogger.get();
        blogger.setId(origin.getId());
        blogger.setCreateTime(origin.getCreateTime());
        blogger.setCreateBy(origin.getCreateBy());
        blogger.setUpdateBy(systemInfo.getSystemInfo());
        blogger.setUpdateTime(System.currentTimeMillis()/1000);
        blogger.setBloggerType(origin.getBloggerType());
        blogger.setBloggerTypeAuditStatus(origin.getBloggerTypeAuditStatus());
        blogger.setTags(origin.getTags());
        blogger.setAiAge(origin.getAiAge());
        blogger.setAiGender(origin.getAiGender());
        if(blogger.getOTotalFavorited()<0){
            blogger.setOTotalFavorited(0);
        }
        bloggerMainRepository.save(blogger);

        growthScoreTime=(System.currentTimeMillis() - growthScoreTime) / 1000;
        // 每天入排行榜播主打卡 todo 打完卡后可以积累或发送到消息队列中批量处理
        if(blogger.getOUid() == 5831967){
            log.info("................{},mysql已更新完成!",blogger.getOUid());
        }
        if(blogger.getOFollowerCount()>=dataConsumerConfig.getFollowerCount() || bloggerMessage.isRanking()){
            if(blogger.getOUid() == 5831967){
                log.info("................{},条件进入到打点逻辑!",blogger.getOUid());
            }
            saveBloggerRecordMain.handle(blogger);
        }
        //用于批量更新到es
//        bloggerMains.add(blogger);
        esSaveBloggerMain.handle(Collections.singletonList(blogger));
        if(bloggerMessage.getRetryTimes()<bloggerMessage.getRetryLimitTimes()){
            log.info(" 重试{}次后成功:uid={}" ,bloggerMessage.getRetryLimitTimes()-bloggerMessage.getRetryTimes(),bloggerMessage.getUid());
        }
        long tcnt=(System.currentTimeMillis() - start) / 1000;

        if(tcnt>dataConsumerConfig.getUpdateBloggerTime()) {
            log.warn("当前播主耗时较长：uid={},region={},awemeSize={},耗时={}s,播主爬取={},增量与分数统计更新={},视频采集与更新={}",uid, region,bloggerAwemes.size(),tcnt,bloggerSpiderTime,growthScoreTime,awemeSpideUpdateTime);
      }
    }
}
